package com.sh.data.engine.domain.integration.offline.runner;

import cn.hutool.extra.spring.SpringUtil;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.constants.Constants;
import com.sh.data.engine.common.enumDefinition.DataXJobTypeEnum;
import com.sh.data.engine.common.enumDefinition.OfflineSyncTaskStatus;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.domain.base.service.LogService;
import com.sh.data.engine.domain.integration.api.record.model.domain.ApiRunningRecordDomain;
import com.sh.data.engine.domain.integration.api.record.service.ApiRunningRecordService;
import com.sh.data.engine.domain.integration.offline.model.domain.OfflineSyncRunningRecordDomain;
import com.sh.data.engine.domain.integration.offline.model.enums.OfflineSyncExecuTypeEnum;
import com.sh.data.engine.domain.integration.offline.service.OfflineSyncService;
import com.sh.data.engine.domain.integration.offline.service.impl.OfflineSyncRunningRecordServiceImpl;
import com.sh.data.engine.infrastructure.config.DataXConfig;
import com.sh.data.engine.infrastructure.config.FileStorageConfiguration;
import lombok.extern.slf4j.Slf4j;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Objects;

@Slf4j
public class DataXRunner implements Runnable {

    private volatile Process process;

    private static final String SKIP = "in ch.qos.logback";

    private OfflineSyncRunningRecordServiceImpl offlineSyncRunningRecordService =
        SpringUtil.getBean(OfflineSyncRunningRecordServiceImpl.class);

    private ApiRunningRecordService apiRunningRecordService =
        SpringUtil.getBean(ApiRunningRecordService.class);

    private OfflineSyncService offlineSyncService = SpringUtil.getBean(OfflineSyncService.class);

    private DataXConfig dataXConfig = SpringUtil.getBean(DataXConfig.class);

    private FileStorageConfiguration fileStorageConfiguration = SpringUtil.getBean(FileStorageConfiguration.class);

    private LogService logService = SpringUtil.getBean(LogService.class);

    private String jsonPath;

    private Long recordId;

    private Integer executeType;

    private Long taskId;

    // 区分离线同步,接口采集
    private DataXJobTypeEnum taskType;

    private Map<Long, DataXRunner> taskCache;

    private int stop = 0;

    public DataXRunner(
        String jsonPath,
        Long recordId,
        Integer executeType,
        Long taskId,
        DataXJobTypeEnum taskType,
        Map<Long, DataXRunner> taskCache) {
        this.jsonPath = jsonPath;
        this.recordId = recordId;
        this.executeType = executeType;
        this.taskId = taskId;
        this.taskType = taskType;
        this.taskCache = taskCache;
    }

    public void stop() {
        if (null != process) {
            process.destroy();
        }
        stop = 1;
    }

    @Override
    public void run() {
        int taskStatus;
        String cmd = dataXConfig.getCmd(jsonPath);
        log.info("cmd:{}", cmd);

        int code = 0;
        if (stop == 1) {
            updateInfoAndLogStatus(OfflineSyncTaskStatus.stopped.getTaskStatus());
            taskCache.remove(recordId);
            return;
        }
        try {
            process = Runtime.getRuntime().exec(cmd);
            read(process.getInputStream(), recordId);
            read(process.getErrorStream(), recordId);
            code = process.waitFor();
        } catch (Exception e) {
            log.error("执行离线同步任务异常：", e);
        }
        if (code == 0) {
            taskStatus = OfflineSyncTaskStatus.success.getTaskStatus();
        } else if (code == 143) {
            taskStatus = OfflineSyncTaskStatus.stopped.getTaskStatus();
        } else {
            taskStatus = OfflineSyncTaskStatus.failure.getTaskStatus();
        }
        if (stop == 1) {
            taskStatus = OfflineSyncTaskStatus.stopped.getTaskStatus();
        }
        updateInfoAndLogStatus(taskStatus);
        // 移除缓存
        taskCache.remove(recordId);
    }

    /**
     * 更新任务状态
     *
     * @param taskStatus 任务状态
     */
    public void updateInfoAndLogStatus(Integer taskStatus) {
        switch (this.taskType) {
            case SYNC: {
                OfflineSyncRunningRecordDomain update =
                    OfflineSyncRunningRecordDomain.builder()
                        .endTime(new Date())
                        .taskStatus(taskStatus)
                        .build();
                update.setId(Long.valueOf(recordId));
                offlineSyncRunningRecordService.updateRunningRecord(update);
                // 周期运行
                if (executeType.equals(OfflineSyncExecuTypeEnum.AUTOMIC.getStatus())) {
                    offlineSyncService.updateOfflineSyncEndStatus(taskId, taskStatus);
                }
                break;
            }
            case APIWS:
            case APIHTTP:
                ApiRunningRecordDomain update =
                    ApiRunningRecordDomain.builder().endTime(new Date()).taskStatus(taskStatus).build();
                update.setId(Long.valueOf(recordId));
                update.setApiId(taskId);
                apiRunningRecordService.saveRecord(update);
//        if (executeType.equals(OfflineSyncExecuTypeEnum.AUTOMIC.getStatus())) {
//          apiHttpService.updateAfterStop(taskId, taskStatus);
//        }
                break;
            default:
                throw new BusinessException("不支持的datax任务类型");
        }
    }

    public void writeLog(List<String> lines, Long id) {
        if (org.apache.commons.collections4.CollectionUtils.isEmpty(lines)) {
            return;
        }
        final String dir = fileStorageConfiguration.getDir();
        String logPath = Constants.LogFileExp.OFFLINE_SYNC_LOG;
        if (!Objects.equals(DataXJobTypeEnum.SYNC, this.taskType)) {
            logPath = Constants.LogFileExp.OFFLINE_SYNC_API_LOG;
        }
        String offlineSyncLog = String.format("%s/log/%s/%s.log", dir, logPath, id);
        logService.write(offlineSyncLog, lines);
    }

    private void read(InputStream in, Long id) {
        new Thread(
            () -> {
                try (BufferedReader reader =
                         new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) {
                    String line;
                    while ((line = reader.readLine()) != null) {
                        if (line.contains(SKIP)) {
                            continue;
                        }
                        List<String> logs = Lists.newArrayList(line);
                        writeLog(logs, id);
                    }
                } catch (IOException e) {
                    log.error(e.getMessage(), e);
                }
            })
            .start();
    }
}
